package cn.itcast.hotel.service.impl;

import cn.itcast.hotel.mapper.HotelMapper;
import cn.itcast.hotel.pojo.Hotel;
import cn.itcast.hotel.pojo.HotelDoc;
import cn.itcast.hotel.pojo.PageResult;
import cn.itcast.hotel.pojo.RequestParams;
import cn.itcast.hotel.service.IHotelService;
import cn.itcast.hotel.utils.HandleResposeUtils;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.functionscore.FunctionScoreQueryBuilder;
import org.elasticsearch.index.query.functionscore.ScoreFunctionBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;

@Service
public class HotelServiceImpl extends ServiceImpl<HotelMapper, Hotel> implements IHotelService {
    @Autowired
    private HotelMapper hotelMapper;
    @Autowired
    private ThreadPoolExecutor myTreadPool;
    @Autowired
    private RestHighLevelClient client;

    /**
     * 获取时间
     */
    @Override
    public void getDoc() {
        Integer allCount = hotelMapper.selectCount(null);
//        十条数据为一页
        Integer pageSize = 10;
//        一共有多少页的处理
        Integer allPage = allCount % pageSize == 0 ? allCount / pageSize : allCount / pageSize + 1;
//        循环处理这每一页的数据
        for (Integer i = 1; i <= allPage; i++) {
//          使用mp的分页获取数据,使用线程池
//            使用构造器传参数给i和pageSize
            RunableTask runableTask = new RunableTask(i, pageSize);
//            等价于每次循环一次就从线程池中拿一个线程去执行
            myTreadPool.submit(runableTask);
        }
    }

    /**
     * 首次使用countDownLatch 添加计算时间的函数
     */
    @Override
    public void getDoc2() {
//        开始时间
        long startTime = System.currentTimeMillis();

        Integer allCount = hotelMapper.selectCount(null);
//        十条数据为一页
        Integer pageSize = 20;
//        一共有多少页的处理
        Integer allPage = allCount % pageSize == 0 ? allCount / pageSize : allCount / pageSize + 1;

        //        1.构建countDownLatch对象，传递一个cdl值，这个值和线程数一样
        CountDownLatch cdl = new CountDownLatch(allPage);
//        循环处理这每一页的数据
        for (Integer i = 1; i <= allPage; i++) {
//          使用mp的分页获取数据,使用线程池
//            使用构造器传参数给i和pageSize
            RunableTask2 runableTask = new RunableTask2(i, pageSize, cdl);
//            等价于每次循环一次就从线程池中拿一个线程去执行
            myTreadPool.submit(runableTask);
        }
//        分线程执行完,就需要唤醒主线程
        try {
            cdl.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
//        计算执行后的时间
        long endTime = System.currentTimeMillis();
        System.out.println("总用时：" + (endTime - startTime) + "毫秒");
    }


    class RunableTask implements Runnable {
        //        经典的构造器传参数
        int i;
        int pageSize;


        public RunableTask(Integer i, Integer pageSize) {
            this.i = i;
            this.pageSize = pageSize;

        }

        @Override
        public void run() {
            IPage<Hotel> pageInfo = new Page<>(i, pageSize);
//          执行分页查询 不需要执行 直接在pageInfo中有数据
//            TODO 为什么不能使用mapper中的，因为MP本身就有基本的sql功能
//            IPage<Hotel> page = hotelService.page(pageInfo);
            IPage<Hotel> page = page(pageInfo);
            List<Hotel> hotels = page.getRecords();
//           处理每一批次的数据(每次处理的请求不一样 所以需要每次的请求创建不一样)
            BulkRequest bulkRequest = new BulkRequest();
            for (Hotel hotel : hotels) {
                HotelDoc hotelDoc = new HotelDoc(hotel);
                bulkRequest.add(new IndexRequest("hotel")
                        .id(hotelDoc.getId().toString())
                        .source(JSON.toJSONString(hotelDoc), XContentType.JSON));
            }
//          一次发送数据请求，bulk
            try {
                client.bulk(bulkRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
//            走到这里就是代表这个线程走完了 就要将cdl的值减
        }
    }

    class RunableTask2 implements Runnable {
        //        经典的构造器传参数
        int i;
        int pageSize;
        CountDownLatch cdl;

        public RunableTask2(Integer i, Integer pageSize, CountDownLatch cdl) {
            this.i = i;
            this.pageSize = pageSize;
            this.cdl = cdl;
        }

        @Override
        public void run() {
            IPage<Hotel> pageInfo = new Page<>(i, pageSize);
//          执行分页查询 不需要执行 直接在pageInfo中有数据
//            TODO 为什么不能使用mapper中的，因为MP本身就有基本的sql功能
//            IPage<Hotel> page = hotelService.page(pageInfo);
            IPage<Hotel> page = page(pageInfo);
            List<Hotel> hotels = page.getRecords();
//           处理每一批次的数据(每次处理的请求不一样 所以需要每次的请求创建不一样)
            BulkRequest bulkRequest = new BulkRequest();
            for (Hotel hotel : hotels) {
                HotelDoc hotelDoc = new HotelDoc(hotel);
                bulkRequest.add(new IndexRequest("hotel")
                        .id(hotelDoc.getId().toString())
                        .source(JSON.toJSONString(hotelDoc), XContentType.JSON));
            }
//          一次发送数据请求，bulk
            try {
                client.bulk(bulkRequest, RequestOptions.DEFAULT);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
//            走到这里就是代表这个线程走完了 就要将cdl的值减
            cdl.countDown();
        }
    }
//==============================================================================================

    /**
     * 酒店的搜索和分页 以及 条件的过滤 我周边的酒店
     *
     * @param requestParams
     * @return
     */
    static final String SEARCH_INDICES = "hotel";


    @Override
    public PageResult searchHotel(RequestParams requestParams) throws IOException {
//        1.创建请求对象
        SearchRequest searchRequest = new SearchRequest(SEARCH_INDICES);
//        2.构建DSL语句
//        2.1 构建DSL语句(有多个请求的过滤 所以需要单独抽取一个方法)

        buildDSLRequest(searchRequest, requestParams);
//        2.2 分页
        Integer page = requestParams.getPage();
        Integer size = requestParams.getSize();
        searchRequest.source()
//                .query(QueryBuilders.matchAllQuery())====添加这个就不能查询成功
//                起始页
                .from((page - 1) * size)
                .size(size);
//        2.3 排序 (普通字段的排序 和 地理位置的排序)=======》添加了距离位置的判断 广告就不会展示前面 先按照距离 分数 再按照广告
       /* String location = requestParams.getLocation();
//        todo 此处的前端处理不了 只能写死地址
        location = "31.03463, 121.61245";
        if (location != null && !"".equals(location)) {
            searchRequest.source().sort(SortBuilders
                    .geoDistanceSort("location", new GeoPoint(location))
                    .order(SortOrder.ASC)
                    .unit(DistanceUnit.KILOMETERS));
        }*/
//        3.发送请求
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
//        4.解析响应结果（需要新建）
//        HandleResposeUtils.handleRespose(response);
        return Analysisesults(response);
    }


    /**
     * 响应结果处理
     *
     * @param response
     * @return
     */
    private PageResult Analysisesults(SearchResponse response) {
        PageResult pageResult = new PageResult();
//        1.获取hits
        SearchHits searchHits = response.getHits();
        long value = searchHits.getTotalHits().value;
//        总条数
        pageResult.setTotal(value);
//        2.解析每个hit
        SearchHit[] hits = searchHits.getHits();
        List<HotelDoc> hotelDocs = new ArrayList<>();
        for (SearchHit hit : hits) {
            String sourceAsString = hit.getSourceAsString();
//          1  获取的每一条数据
            HotelDoc hotelDoc = JSON.parseObject(sourceAsString, HotelDoc.class);
//          2  获取距离的排序字段 distance（和source同级）
            Object[] sortValues = hit.getSortValues();
            if (sortValues.length > 0) {
                Object sortValue = sortValues[0];
//                将值设置进去
                hotelDoc.setDistance(sortValue);
            }
//            3.高亮部分的解析
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();
            if (!CollectionUtils.isEmpty(highlightFields)) {
//                todo 此处的all的字段高亮如何解析 拼接
//                HighlightField all = highlightFields.get("all");
                HighlightField all = highlightFields.get("name");
//                判断all是否为非空
                if (all != null) {
                    String fragment = all.getFragments()[0].toString();
                    hotelDoc.setName(fragment);
                }
            }
            hotelDocs.add(hotelDoc);
        }
        pageResult.setHotels(hotelDocs);
        return pageResult;
    }


    /**
     * 布尔多个条件的过滤操作boolQuery
     */
    private void buildDSLRequest(SearchRequest searchRequest, RequestParams requestParams) throws IOException {
//        多个组合条件需要使用布尔查询,并且需要判断过滤条件和关键字是否为空

        String key = requestParams.getKey();
//        1.构建DSL语句===多条件需要使用布尔查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//        todo 关键
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
//        1.1关键字的搜索
        if (key == null || "".equals(key)) {
            boolQueryBuilder.must(QueryBuilders.matchAllQuery());
        } else {
//           添加了关键字的搜索(可以进行高亮的处理)
//            boolQueryBuilder.must(QueryBuilders.matchQuery("all", key));
            boolQueryBuilder.must(QueryBuilders.matchQuery("all", key));
//            高亮显示
            HighlightBuilder highlightBuilder = new HighlightBuilder();
//            highlightBuilder.field("all").requireFieldMatch(false);
            highlightBuilder.field("name").requireFieldMatch(false);
            searchSourceBuilder.highlighter(highlightBuilder);
        }
//        1.2.品牌的过滤 term
        String brand = requestParams.getBrand();
//        相比拿brand去安全不会出现空指针异常
        if (brand != null && !"".equals(brand)) {
            boolQueryBuilder.filter(QueryBuilders.termQuery("brand", brand));
        }
//        1.3星级过滤 term
        String starName = requestParams.getStarName();
        if (starName != null && !"".equals(starName)) {
            boolQueryBuilder.filter(QueryBuilders.termQuery("starName", starName));
        }
//        1.4价格 range
        Integer maxPrice = requestParams.getMaxPrice();
        Integer minPrice = requestParams.getMinPrice();
        if (maxPrice != null && minPrice != null) {
//            大于等于min 小于max
            boolQueryBuilder.filter(QueryBuilders.rangeQuery("price").gte(minPrice).lt(maxPrice));
        }
//        1.5城市 term
        String city = requestParams.getCity();
        if (city != null && !"".equals(city)) {
            boolQueryBuilder.filter(QueryBuilders.termQuery("city", city));
        }
//        todo 将布尔放入是query 不是布尔
        searchSourceBuilder.query(boolQueryBuilder);

//            2.将构建好的DSL语句放入source进行发送
//        searchRequest.source().query(boolQueryBuilder);
        /**
         *  算分函数 测试需要先把距离排序的代码注释 否则会影响
         */

//        3.添加算分函数(需要把布尔查询的包裹起来 一起发给source)
        FunctionScoreQueryBuilder functionScoreQueryBuilder = QueryBuilders.functionScoreQuery(
//                  原始查询，相关性算分的查询 searchRequest.source().query(boolQueryBuilder);
//  todo          此处的加入布尔查询的结果不是绑定 一起发送请求 而是将布尔查询的结果参与算分
                boolQueryBuilder,
                // function score的数组
                new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{
//                其中一个function score 元素
                        new FunctionScoreQueryBuilder.FilterFunctionBuilder(
                                // 过滤条件
                                QueryBuilders.termQuery("isAD", true),
                                // 算分函数
                                ScoreFunctionBuilders.weightFactorFunction(10)
                        )
                }
        );
//         2.将构建好的DSL语句放入source进行发送
//        searchRequest.source().query(functionScoreQueryBuilder);
//        将布尔查询 高亮 算分一起放入searchSourceBuilder中 发送请求
        searchSourceBuilder.query(functionScoreQueryBuilder);
        searchRequest.source(searchSourceBuilder);
    }

    /**
     * 使用聚合功能，利用Bucket聚合，对搜索结果中的文档基于品牌分组、基于城市分组，就能得知包含哪些品牌、哪些城市了。
     */
    @Override
    public Map<String, List<String>> filtersHotel(RequestParams requestParams) {
        try {
//        1.创建请求
            SearchRequest searchRequest = new SearchRequest(SEARCH_INDICES);
//        2.DSL语句的查询过滤query 和list一样
            buildDSLRequest(searchRequest, requestParams);
//        2.DSL语句 聚合
//            // 2.2.设置size
//        request.source().size(0);
            searchRequest.source().size(0);
//            聚合结果
            buildResult(searchRequest);
//        3.发送请求
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
//        4.解析结果
//        4.解析结果
            Map<String, List<String>> map = new HashMap<>();
            Aggregations aggregations = response.getAggregations();
            List<String> city = handleMap(aggregations, "city_agg");
            map.put("城市", city);
            List<String> starName = handleMap(aggregations, "starName_agg");
            map.put("星级", starName);
            List<String> brand = handleMap(aggregations, "brand_agg");
            map.put("品牌", brand);
            return map;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 聚合DSL
     *
     * @param searchRequest
     */
    private void buildResult(SearchRequest searchRequest) {
//        3.city
        searchRequest.source().aggregation(
                AggregationBuilders.terms("city_agg").field("city").size(100)
        );
//        2.star
        searchRequest.source().aggregation(
                AggregationBuilders.terms("starName_agg").field("starName").size(100)
        );
//        1.brand
        searchRequest.source().aggregation(
                AggregationBuilders.terms("brand_agg").field("brand").size(100)
        );
    }

    /**
     * 解析结果
     *
     * @param aggregations
     * @param aggName
     * @return
     */
    private static List<String> handleMap(Aggregations aggregations, String aggName) {
        // 4.1.根据聚合名称获取聚合结果
        Terms aggregation = aggregations.get(aggName);
        // 4.2.获取buckets
        List<? extends Terms.Bucket> buckets = aggregation.getBuckets();
        List<String> list = new ArrayList<>();
        for (Terms.Bucket bucket : buckets) {
            String keyAsString = bucket.getKeyAsString();
            list.add(keyAsString);
        }
        return list;
    }


    @Override
    public Map<String, List<String>> filters(RequestParams requestParams) {
        try {
            // 1.准备Request
            SearchRequest request = new SearchRequest("hotel");
            // 2.准备DSL
            // 2.1.query
            buildDSLRequest(request, requestParams);
            // 2.2.设置size
            request.source().size(0);
            // 2.3.聚合
            buildAggregation(request);
            // 3.发出请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 4.解析结果
            Map<String, List<String>> result = new HashMap<>();
            Aggregations aggregations = response.getAggregations();
            // 4.1.根据品牌名称，获取品牌结果
            List<String> brandList = getAggByName(aggregations, "brandAgg");
            result.put("brand", brandList);
            // 4.2.根据品牌名称，获取品牌结果
            List<String> cityList = getAggByName(aggregations, "cityAgg");
            result.put("city", cityList);
            // 4.3.根据品牌名称，获取品牌结果
            List<String> starList = getAggByName(aggregations, "starAgg");
            result.put("starName", starList);

            return result;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    private void buildAggregation(SearchRequest request) {
        request.source().aggregation(AggregationBuilders
                .terms("brandAgg")
                .field("brand")
                .size(100)
        );
        request.source().aggregation(AggregationBuilders
                .terms("cityAgg")
                .field("city")
                .size(100)
        );
        request.source().aggregation(AggregationBuilders
                .terms("starAgg")
                .field("starName")
                .size(100)
        );
    }

    private List<String> getAggByName(Aggregations aggregations, String aggName) {
        // 4.1.根据聚合名称获取聚合结果
        Terms brandTerms = aggregations.get(aggName);
        // 4.2.获取buckets
        List<? extends Terms.Bucket> buckets = brandTerms.getBuckets();
        // 4.3.遍历
        List<String> brandList = new ArrayList<>();
        for (Terms.Bucket bucket : buckets) {
            // 4.4.获取key
            String key = bucket.getKeyAsString();
            brandList.add(key);
        }
        return brandList;
    }

    /**
     * 自动补全的接口
     */
    @Override
    public List<String> suggestion(String prefix) {
        try {
//        1.准备请求
            SearchRequest searchRequest = new SearchRequest(SEARCH_INDICES);
//        2.DSL语句的suggestion查询
//        2.1 查询的集合名称
            String SuggestName = "SuggestName";
            searchRequest.source()
                    .suggest(new SuggestBuilder().addSuggestion(
                            SuggestName,
                            SuggestBuilders.completionSuggestion("suggestion")
                                    //                        前缀
                                    .prefix(prefix)
                                    //                                是否跳过重复的
                                    .skipDuplicates(true)
                                    .size(10)


                    ));
//        3.发送请求
            SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
//        4.解析结果返回
            Suggest suggest = response.getSuggest();
            CompletionSuggestion suggestion = suggest.getSuggestion(SuggestName);
            List<CompletionSuggestion.Entry.Option> options = suggestion.getOptions();
            List<String> strings = new ArrayList<>();
            for (CompletionSuggestion.Entry.Option option : options) {
                String string = option.getText().toString();
                strings.add(string);
            }
            return strings;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
